package com.sh.data.engine.domain.datadev.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sh.data.engine.common.constants.MapReduceConstants;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.common.service.impl.CommandExecutorServiceImpl;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.service.DatadevResourceService;
import com.sh.data.engine.domain.datadev.offline.service.DatadevScriptService;
import com.sh.data.engine.domain.datadev.offline.service.FreemarkerService;
import com.sh.data.engine.domain.shims.hdfs.util.HdfsUtil;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevJobEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevMenuNodeEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevResourceEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevScriptEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.exec.LogOutputStream;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.joda.time.DateTime;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Slf4j
public class OfflineDevMapReduceTaskJobRunner extends OfflineDevBaseJobRunner {

    private static final Splitter SPLITTER = Splitter.on(Pattern.compile("\\s+"));
    private static final Joiner JOINER = Joiner.on(" ").skipNulls();

    private DatadevResourceService datadevResourceService =
        SpringUtil.getBean(DatadevResourceService.class);
    private FreemarkerService freemarkerService = SpringUtil.getBean(FreemarkerService.class);
    private CommandExecutorServiceImpl commandExecutorService =
        SpringUtil.getBean(CommandExecutorServiceImpl.class);
    private UserService userService = SpringUtil.getBean(UserService.class);
    private DatadevScriptService datadevScriptService =
        SpringUtil.getBean(DatadevScriptService.class);

    public OfflineDevMapReduceTaskJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService) {
        super(
            jobEntity,
            menuNodeEntity,
            parentNodeIdAndLineStatusMap,
            childNodeIdList,
            allJobNodeMap,
            executorService);
    }

    @Override
    public void cancel() {
        // todo
    }

    @Override
    public int runJob() throws Exception {

        File jarFile = null;
        try {
            // 检查语句是否合法
            DatadevScriptEntity taskConfigEntity =
                datadevScriptService.getDatadevScriptByNodeId(menuNodeEntity.getId());
            Long projectId = menuNodeEntity.getProjectId();
            String jarCmd = taskConfigEntity.getScript();

            String cmd = validateJarCmd(jarCmd);

            // 下载jar
            jarFile = downloadJar(cmd, projectId);

            // 生成shell
            String mapreduceExecuteShell = mapReduceConfiguration.getMapreduceExecuteShell();
            String shell = generateShell(cmd, jarFile, mapreduceExecuteShell);

            int code = executeShell(shell);
        } catch (Exception e) {
            log.error(e.getMessage());
            return OfflineDevJobStatusEnum.FAILURE.getCode();
        }

        return OfflineDevJobStatusEnum.SUCCESS.getCode();
    }

    private int executeShell(String shell) {
        // 将shell内容写出到指定shell路径下，注意设置执行权限
        File shellFile = generateTaskFile(shell);
        if (shellFile == null) {
            throw new BusinessException("the mapreduce shell file couldn't be created!");
        }

        UserDomain userDomain = userService.getUserById(jobEntity.getCreatorId());
        int code =
            commandExecutorService.executeShell(
                shellFile.getAbsolutePath(), userDomain.getCmdUser(), createLogOutStream());

        return code;
    }

    private File generateTaskFile(String content) {

        DateTime execDt = new DateTime(System.currentTimeMillis());
        String runMonth = execDt.toString("yyyyMM");
        String runTime = execDt.toString("yyyyMMddHHmmss");
        String fileName = String.format("%s-%s.sh", jobEntity.getId(), runTime);
        String mapreduceScriptDir = mapReduceConfiguration.getMapreduceScriptDir();
        File taskFile = null;
        try {
            String fileDir =
                StringUtils.isBlank(mapreduceScriptDir)
                    ? System.getProperty("user.dir") + File.separator + "tasks"
                    : mapreduceScriptDir;
            File dir = new File(fileDir, runMonth);
            if (!dir.exists()) {
                dir.mkdirs();
            }

            taskFile = new File(dir, fileName);
            FileUtils.write(taskFile, content, StandardCharsets.UTF_8.name(), false);
            taskFile.setExecutable(true, false);
        } catch (IOException e) {
            log.error("generate task file [{}] error: ", fileName, e);
        }
        return taskFile;
    }

    private String generateShell(String jarCmd, File jarFile, String shellFile) {
        // 替换原始命令中jar包路径为当前运行环境下实际jar包的完整路径
        String jarFilePath = jarFile.getAbsolutePath();
        jarFilePath = String.format("jar %s", jarFilePath);
        jarCmd = jarCmd.replaceFirst(MapReduceConstants.P_JAR.pattern(), jarFilePath);
        Map<String, Object> values = Maps.newHashMap();
        values.put(MapReduceConstants.HADOOP_JAR_CMD, jarCmd);
        String mapreduceExecuteShell = mapReduceConfiguration.getMapreduceExecuteShell();
        ;
        String content = null;
        try {
            content = freemarkerService.present(shellFile, values);
        } catch (Exception e) {
            throw new BusinessException("process shell template [" + mapreduceExecuteShell + "] error!");
        }
        return content;
    }

    private String validateJarCmd(String jarCmd) {
        if (StringUtils.isBlank(jarCmd)) {
            return null;
        }

        jarCmd = StringUtils.endsWith(jarCmd, ";") ? jarCmd.replace(";", "") : jarCmd;

        Matcher m = MapReduceConstants.P_HADOOP_JAR.matcher(jarCmd);
        if (!m.matches()) {
            return null;
        }
        /**
         * 对hadoop jar命令行做进一步的处理： 1 不允许在命令行里面指定配置文件路径，如果指定了需要删除掉。 2 如果用户自己指定了诸如参数“-libjars
         * $LIB_JARS”，需要替换为当前环境下的值。 3 如果hadoop jar命令带有自定义参数，需要处理自定义参数与系统参数-libjars的顺序
         */
        String configSlice = m.group(1), libSlice = m.group(2);
        if (StringUtils.isNotBlank(configSlice)) {
            jarCmd = StringUtils.trim(StringUtils.remove(jarCmd, configSlice));
        }
        if (StringUtils.isNotBlank(libSlice)) {
            jarCmd = StringUtils.trim(StringUtils.remove(jarCmd, libSlice));
        }
        List<String> slices = Lists.newLinkedList(SPLITTER.splitToList(jarCmd));
        String jarSlice =
            slices.stream().filter(e -> StringUtils.endsWithIgnoreCase(e, ".jar")).findFirst().get();
        int jarPos = slices.indexOf(jarSlice);
        if (jarPos + 2 < slices.size()) {
            slices.add(jarPos + 2, MapReduceConstants.ENV_VAR_LIBJARS_KEY);
            slices.add(jarPos + 3, MapReduceConstants.ENV_VAR_LIBJARS_VAL);
        }
        if (slices.indexOf(MapReduceConstants.ENV_VAR_LIBJARS_KEY) == -1) {
            slices.add(MapReduceConstants.ENV_VAR_LIBJARS_KEY);
            slices.add(MapReduceConstants.ENV_VAR_LIBJARS_VAL);
        }
        jarCmd = JOINER.join(slices);
        return jarCmd;
    }

    private File downloadJar(String jarCmd, Long projectId) {
        // 从命令行抽取出jar包名称
        if (StringUtils.isBlank(jarCmd)) {
            throw new BusinessException("the hadoop jar cmd is empty!");
        }
        Matcher m = MapReduceConstants.P_JAR.matcher(jarCmd);
        String jarName = null;
        if (m.find()) {
            jarName = m.group(1);
            jarName = jarName.replaceFirst("jar\\s+", "");
        }
        if (StringUtils.isBlank(jarName)) {
            throw new BusinessException(
                "the jar name within cmd[" + jarCmd + "] illegal!");
        }

        // 从HDFS中下载jar包到本地文件系统路径下
        DatadevResourceEntity jarResource =
            datadevResourceService.findByProjectIdAndName(projectId, jarName);
        if (jarResource == null) {
            throw new BusinessException(
                "the jar [" + jarName + "] with projectId [" + projectId + "] not exist in db!");
        }
        String hdfsPath = jarResource.getFilePath();
        String username = MapReduceConstants.HDFS_USERNAME;
        String localJarName = String.format("%d_%s", projectId, jarName);
        String mapreduceJarDir = mapReduceConfiguration.getMapreduceJarDir();
        String hdfsUrl = hdfsConfig.getHdfsUrl();
        String hdfsNamenodes = hdfsConfig.getHdfsNamenodes();
        String hdfsNamenodesaddr = hdfsConfig.getHdfsNamenodesaddr();
        String hdfsNameservices = hdfsConfig.getHdfsNameservices();
        File target = new File(new File(mapreduceJarDir), localJarName);
        try (FileSystem hdfs =
                 HdfsUtil.getFileSystem(
                     null, username, null, hdfsUrl, hdfsNameservices, hdfsNamenodesaddr, hdfsNamenodes)) {
            Path src = new Path(hdfsPath);
            Path dst = new Path(new URI(target.getAbsolutePath()));
            if (hdfs.exists(src)) {
                FileUtils.forceMkdirParent(target);
                hdfs.copyToLocalFile(src, dst);
            }
        } catch (IOException | URISyntaxException e) {
            if (target != null && target.exists()) {
                target.delete();
            }
            throw new BusinessException(
                "download hdfs jar error: jarCmd is [" + jarCmd + "], projectId is [" + projectId + "].");
        }

        return (target.exists() && target.isFile() && target.length() > 0) ? target : null;
    }

    private LogOutputStream createLogOutStream() {

        return new LogOutputStream() {
            @Override
            protected void processLine(String line, int logLevel) {
                writeLog(Lists.newArrayList(line));
            }
        };
    }
}
